package com.galeno.demo;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/9/119:30
 */
public class ConsumerOffsetTest {
    public static void main(String[] args) {
        String serverUrl = "galeno01:9092,galeno02:9092,galeno03:9092";
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, serverUrl);
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 如果没有找到之前记录的偏移量，就从这个参数指定的位置开始消费
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 消费者的唯一标识
        properties.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "c01");
        // 消费者所属的组id
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "g01");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

        TopicPartition tpc3_p0 = new TopicPartition("first", 0);
        //客户端由消费者固定住要消费的topic以及分区
        consumer.assign(Collections.singletonList(tpc3_p0));
        // seek定位到指定的偏移量开始消费（无视之前所记录的消费偏移量）
        consumer.seek(tpc3_p0,3);
        while(true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            if(!records.isEmpty()){
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(record.partition() + " -> " + record.offset() + " -> " + record.value());
                }

            }
        }

    }
}
